package com.mini.gateway.core.modules.common;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * KafkaService
 *
 */
@Slf4j
@Component
public class KafkaService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 注入KafkaTemplate
     * @param kafkaTemplate kafka模版类
     */
    @Autowired
    public KafkaService(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String topic, String key, String data) {
        try {
            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, data);
            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable ex) {
                    log.error("【kafka sendMessage】error, ex = {}, topic = {}, data = {}", ex, topic, data);
                }

                @Override
                public void onSuccess(SendResult<String, String> result) {
                    log.info("【kafka sendMessage】success topic = {}, data = {}",topic, data);
                }
            });
        } catch (Exception ex) {
            log.error("【kafka sendMessage】Exception", ex);
        }

    }
}
